/**
 * @author zhounan 
 * @date 2024年7月3日
 * @version V1.0
 */
package org.zn.quality.handler;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.commons.lang3.tuple.Pair;
import org.springframework.stereotype.Component;
import org.zn.etl.config.Constant;
import org.zn.etl.misc.CompareUtils;
import org.zn.quality.entity.QualityStrategyEntity;
import org.zn.quality.entity.po.QualityStatisticsColumnRulePo;

import com.google.common.collect.Maps;

/**
 * @author zhounan 
 * @date 2024年7月3日
 * @version V1.0
 */
@Component("MapCompareHandler")
public class MapCompareHandler  implements QualityHandler<Map<String, Object>, Map<String, Object>> {
	


	@Override
	public Map<String, Object> doHandler(List<Map<String, Object>> srcData, List<Map<String, Object>> tagData,
			QualityStrategyEntity strategy)  {
		// TODO Auto-generated method stub
		
		Map<String, Object> inspectResult =new HashMap<>();
		
		List<QualityStatisticsColumnRulePo>   columnRules=strategy.getStatisticsColumnRule();

		List<QualityStatisticsColumnRulePo> keys=columnRules.stream().filter(it-> it.getOperator().equals("key")).collect(Collectors.toList());

		QualityStatisticsColumnRulePo skeyCol=keys.stream().filter(it-> it.getSide().equals(Constant.SOURCE)).findAny().get();
		QualityStatisticsColumnRulePo tkeyCol=keys.stream().filter(it-> it.getSide().equals(Constant.TARGET)).findAny().get();
		
		List<QualityStatisticsColumnRulePo> valueCol=columnRules.stream().filter(it-> it.getOperator().equals("value")).collect(Collectors.toList());
		
		Map<Object,Map<String, Object>>  srcMd= srcData.stream().collect(Collectors.toMap( it-> it.get(skeyCol.getColumn().toLowerCase()), Function.identity()));
		Map<Object,Map<String, Object>>  tatMd= tagData.stream().collect(Collectors.toMap( it->  it.get(tkeyCol.getColumn().toLowerCase()) , Function.identity()));
		
		List<Pair<Map<String, Object>, Map<String, Object>>> diff=new ArrayList<>();   
		
		for (Entry<Object,Map<String, Object>> srcRow : srcMd.entrySet()) {
			
			Object  key=srcRow.getKey();
			
			Map<String, Object>  srow=srcRow.getValue();
			Map<String, Object>  trow=tatMd.get(key);
			
			for (QualityStatisticsColumnRulePo cr : valueCol) {
				
				Object so=srow.get(cr.getColumn());
				
				Object to=trow.get(cr.getMappingName()!=null?cr.getMappingName():cr.getColumn());
				
				if(!  CompareUtils.valueEquals(so, to) ) {
					diff.add(Pair.of(srow, trow));
					break;
				}	
				
			}	
			
		}
		
		
		Map<String, Object> datas= Maps.newHashMap();
		datas.put("srcData", srcData);
		datas.put("tagData", tagData);
		inspectResult.put("datas", datas);

		inspectResult.put(Constant.DIFF, diff);
		inspectResult.put(Constant.DIFF_SIZE, diff.size());
		
		return inspectResult;
	}
	

}
